package com.bigdata.spark.core.wc

import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import scala.collection.mutable
import org.apache.spark.sql.functions._

/**
 * https://blog.csdn.net/m0_37813354/article/details/103981121?utm_medium=distribute.pc_relevant_t0.none-task-blog-BlogCommendFromBaidu-1.control&depth_1-utm_source=distribute.pc_relevant_t0.none-task-blog-BlogCommendFromBaidu-1.control
 * 我们遇到这样一个问题，
 * 1 先生成一个包含所有预置字段的DataFrame（此时的字段名称是英文的）
 * 2 根据传入的条件动态组合字段从1中的大的DataFrame 里 select的到想要的字段
 * 3 但是我想存入的csv文件中要有中文表头
 * 4 遇到的问题是：使用sql 拼接然后as 取别名为中文，会报出不支持
 * 所以想到如下的方法来解决：
 */
object SplitDFChinese {

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .appName("Example")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._
    //第一步：的到所有字段的DataFrame
    val df: DataFrame = Seq(
      ("1001", "张三", 50),
      ("1002", "李四", 20),
      ("1003", "王五", 30),
      ("1004", "马六", 70),
      ("1005", "赵四", 20),
      ("1006", "刘五", 30),
      ("1007", "陈六", 70),
      ("1008", "朱四", 20),
      ("1009", "杨五", 30),
      ("1010", "吴六", 70),
      ("1011", "严四", 20),
      ("1012", "马五", 30),
      ("1013", "赵六", 70)
    ).map(r=>(r._1,r._2,new Integer(r._3))).toDF("id","name","age")

    //第二步：准备英文中文对照字典表，放map里
    val map: mutable.Map[String, String] = mutable.Map("id"->"编号",
      "name"->"名字",
      "age"->"年龄")

    // 第三步：计算结果注册成一个临时表
    df.createOrReplaceTempView("my_table_test")
    //第四步：我的动态拼接的字段和sql
    val mySql = "select id, name from my_table_test"
    //第五步：可以写入csv文件的所有需要的字段和值
    val englishColumnsDF = spark.sql(mySql)
    englishColumnsDF.printSchema()
    englishColumnsDF.show()
    //第六步：调用我自己定义的多列重命名方法，把应为列名转换成中文列名
    val chinestColumnDF = columnNamesRename(englishColumnsDF,map)
    chinestColumnDF.show()
  }

  /**
   * 多列同时重命名
   * @param df 传入要重命名的DataFrame
   * @param map 传入预先准备好的英文中文对照字典表
   * @return 返回重新命名后的DataFrame
   */
  def columnNamesRename(df:DataFrame, map: mutable.Map[String, String]) = {
    val oldColumnsNames:Seq[String] = df.columns.toSeq
    val renameColumns:Seq[Column] = oldColumnsNames.map(name => {
      val newName = map.get(name).getOrElse("")
      col(name).as(newName)
    })
    df.select(renameColumns:_*)
  }

  /**
   * 传入DataFrame 将schema转换为大写
   * @param df
   * @return
   */
  def columnNamesRename(df:DataFrame) = {
    val oldColumnsNames:Seq[String] = df.columns.toSeq
    val renameColumns:Seq[Column] = oldColumnsNames.map(name => {
      val newName = name.toUpperCase
      col(name).as(newName)
    })
    df.select(renameColumns:_*)
  }
}
